package org.aktivezone;

import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

import org.aktivezone.message.handling.PayloadHandler;
import org.aktivezone.message.serializer.Serializer;
import org.aktivezone.zone.DirectZone;
import org.aktivezone.zone.RoutingStrategy;
import org.aktivezone.zone.registry.ZoneRegistry;

import com.google.common.base.Stopwatch;

public class AdaApp //implements AdaAppInterface 
{
	private static final int second = 1000;
	private static final String STOP_MSG = "STOP";
	private static final String RK = "String";
	private static final byte[] STOP = STOP_MSG.getBytes();
	private String iface = "10.6.67.54";
	private final String zoneName= "ADA-LivingRoom";
	private final int numberOfThreads = 1;
	private static int port= 5556;
	private final int batchSize = 1024;
	private final String payload = "a";
	private final RoutingStrategy strategy = RoutingStrategy.FAN_OUT;
	private ZoneRegistry registry;
	private DirectZone zone;
	private Stopwatch collectWatch = new Stopwatch();
	private Stopwatch emitWatch = new Stopwatch();
	private CountDownLatch emitLatch;
	private static final String ROUTING_KEY = "RK0";
	
public static void main(String[] args) {

	 port = Integer.parseInt(args[0]);	
AdaApp ada = new AdaApp();
try {
	ada.runIns();
	Runtime.getRuntime().addShutdownHook(new ShutdownThread(ada));
	//ada.stop();
} catch (Exception e) {
	// TODO Auto-generated catch block
	e.printStackTrace();
}
}

private void runIns() throws IOException, InterruptedException {
	final AtomicBoolean receiving = new AtomicBoolean(false);
	registry = new ZoneRegistry();
	registry.setLocalhost(iface);
	registry.setZoneName(zoneName);
	registry.setPort(port);
	registry.start();
	zone = new DirectZone(registry, numberOfThreads);
	zone.bind(new Serializer<String>() {

		@Override
		public byte[] serialize(String payload) {
			return payload.getBytes();
		}

		@Override
		public String deserialize(byte[] bytes) {
			return new String(bytes);
		}

		@Override
		public Class<String> getHandledType() {
			return String.class;
		}
	}, new PayloadHandler<String>() {

		@Override
		public void handle(String m) {
		System.out.println("received:"+m);
			if (receiving.compareAndSet(false, true)) {
				collectWatch.start();
			}

		}
	});
	zone.start();
	

	
		zone.emit(payload, ROUTING_KEY, RoutingStrategy.FAN_OUT);

//		zone.emit(payload, ROUTING_KEY, RoutingStrategy.ROUND_ROBIN);
//
//		zone.emit(payload, ROUTING_KEY, RoutingStrategy.ROUND_ROBIN);

	System.gc(); 
}

private static final class ShutdownThread extends Thread {

	private final AdaApp m;

	private ShutdownThread(AdaApp mr) {
		super();
		this.m = mr;
	}

	@Override
	public void run() {
		m.stop();
	}
};

private void printStats(String stat, int batchSize, Stopwatch watch) {
	long ms = watch.elapsedMillis();
	double tpt = ((double) batchSize) / ms * second;
	System.out.printf("%n%-20s%n", stat); 
	System.out.println("----------------------------------------------------------------------------------");
	System.out.printf("%-20s %-28s %-31s%n", "Total msgs", "Elapsed Time (ms)", "Throughput (msgs/s)");
	System.out.println("----------------------------------------------------------------------------------");
	System.out.printf("%15d %21d %30.3f%n", batchSize, ms, tpt);
	System.out.println("-----------------------------------------------------------------------------------");
	watch.reset();
}

public void stop() {
	zone.close();
	registry.close();
}
}

